package drds.binlog.parse.index;

import drds.binlog.common.position.LogPosition;
import drds.binlog.parse.exception.BinlogParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


public class PeriodMixedLogPositionManager extends AbstractLogPositionManager
{

    private static final Logger logger = LoggerFactory.getLogger(PeriodMixedLogPositionManager.class);
    @SuppressWarnings("serial")
    private final LogPosition nullPosition = new LogPosition()
    {
    };
    private MemoryLogPositionManager memoryLogPositionManager;
    private ZooKeeperLogPositionManager zooKeeperLogPositionManager;
    private ScheduledExecutorService scheduledExecutorService;
    private long period;
    private Set<String> persistTasks;

    public PeriodMixedLogPositionManager(MemoryLogPositionManager memoryLogPositionManager,
                                         ZooKeeperLogPositionManager zooKeeperLogPositionManager, long period)
    {
        if (memoryLogPositionManager == null)
        {
            throw new NullPointerException("null memoryLogPositionManager");
        }

        if (zooKeeperLogPositionManager == null)
        {
            throw new NullPointerException("null zooKeeperLogPositionManager");
        }

        if (period <= 0)
        {
            throw new IllegalArgumentException("period must be positive, given: " + period);
        }

        this.memoryLogPositionManager = memoryLogPositionManager;
        this.zooKeeperLogPositionManager = zooKeeperLogPositionManager;
        this.period = period;
        this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
        this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
    }

    @Override
    public void stop()
    {
        super.stop();

        if (zooKeeperLogPositionManager.isRunning())
        {
            zooKeeperLogPositionManager.stop();
        }

        if (memoryLogPositionManager.isRunning())
        {
            memoryLogPositionManager.stop();
        }

        scheduledExecutorService.shutdown();
    }

    @Override
    public void start()
    {
        super.start();

        if (!memoryLogPositionManager.isRunning())
        {
            memoryLogPositionManager.start();
        }

        if (!zooKeeperLogPositionManager.isRunning())
        {
            zooKeeperLogPositionManager.start();
        }

        // 启动定时工作任务
        scheduledExecutorService.scheduleAtFixedRate(new Runnable()
        {

            public void run()
            {
                List<String> tasks = new ArrayList<String>(persistTasks);
                for (String destination : tasks)
                {
                    try
                    {
                        // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
                        zooKeeperLogPositionManager.persistLogPosition(destination, getLatestIndexBy(destination));
                        persistTasks.remove(destination);
                    } catch (Throwable e)
                    {
                        // ignore
                        logger.error("period update" + destination + " curosr failed!", e);
                    }
                }
            }
        }, period, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public LogPosition getLatestIndexBy(String destination)
    {
        LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
        if (logPosition == nullPosition)
        {
            return null;
        } else
        {
            return logPosition;
        }
    }

    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws BinlogParseException
    {
        persistTasks.add(destination);
        memoryLogPositionManager.persistLogPosition(destination, logPosition);
    }
}
